package com.lemon.rpcframe.consumer.business;

import java.lang.reflect.Method;
import java.util.Date;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.log4j.Logger;

import com.alibaba.fastjson.JSON;
import com.google.common.base.Preconditions;
import com.google.common.eventbus.EventBus;
import com.google.common.eventbus.Subscribe;
import com.google.inject.Singleton;
import com.lemon.rpcframe.annotation.Business;
import com.lemon.rpcframe.consumer.proxy.JDKProxy;
import com.lemon.rpcframe.contract.RemoteTaskService;
import com.lemon.rpcframe.dao.entity.TaskExecutableBean;
import com.lemon.rpcframe.dao.entity.TaskLogBean;
import com.lemon.rpcframe.provider.Damon;
import com.lemon.rpcframe.util.ClassHelper;
import com.lemon.rpcframe.util.GenericsUtils;
import com.lemon.rpcframe.util.PackageScan;
import com.lemon.rpcframe.util.StringUtils;

/**
 * 拉取任务  并执行
 * 1 定时任务  每个Task都可以执行
 * 2 实时任务  只有Master节点可以执行    暂不做 
 * 3 任务执行完 一定要返回一个结果  以便在页面中明显的看出来
 * 4 任务的执行入口在 taskdefine.mainClass中定义   一个
 * 
 * 
 * @author wangyazhou
 * @version 1.0 
 * @date  2015年9月18日 下午3:36:10
 * @see 
 * @since
 */
@Singleton
public class TaskPullExecutor implements Damon {

    private static final Logger logger = Logger.getLogger(TaskPullExecutor.class);
    private final ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);
    private final ConcurrentHashMap<String, Object> businessMap = GenericsUtils.newConcurrentHashMap();
    private final ConcurrentHashMap<String, Class<?>> classMap = GenericsUtils.newConcurrentHashMap();
    private final ReentrantLock initLock = new ReentrantLock();
    private EventBus taskBus;

    @Override
    public void start() throws Exception {
        try {
            findAll();
            taskBus = new EventBus("task");
            taskBus.register(new BusinessExecutor());
            executor.scheduleWithFixedDelay(new Runnable() {
                @Override
                public void run() {
                    try {
                        RemoteTaskService taskService = JDKProxy.createProxy(RemoteTaskService.class, "remote");
                        TaskExecutableBean task = taskService.taskPull("test", "test-demo");
                        //远程获取待执行的任务 并义给对事的事件机制执行
                        if (task != null) {
                            logger.info(JSON.toJSONString(task));
                            taskBus.post(task);
                        } else {
                            logger.info(String.format("pulltask result is null "));
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }, 5, 10, TimeUnit.SECONDS);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void stop() throws Exception {
        try {
            executor.shutdown();
        } catch (Exception e) {//防御性捕获
            e.printStackTrace();
        }
    }

    /**
     * 获取所有带business注解的类
     * 要求同一个类名 不有对应同一个Class
     */
    private void findAll() {
        try {
            Set<Class<?>> allClass = PackageScan.scanClasses("com.bj58.zptask");
            for (Class<?> cls : allClass) {
                if (cls.isAnnotationPresent(Business.class)) {
                    String name = cls.getName();
                    classMap.put(name.substring(name.lastIndexOf(".") + 1, name.length()), cls);
                }
            }
        } catch (Exception e) {

        }
    }

    private class BusinessExecutor {
        //执行一个任务
        @Subscribe
        public void doExecutableTask(TaskExecutableBean etask) {
            Preconditions.checkNotNull(etask);
            if (StringUtils.isNullOrEmpty(etask.getMainClass())) {
                return;
            }
            try {
                Class<?> mainClass = ClassHelper.forName(etask.getMainClass());
                if (TaskBusiness.class.isAssignableFrom(mainClass)) {
                    //需保证每一个类只能被初始化一次
                    initLock.tryLock(1000, TimeUnit.MILLISECONDS);
                    Object obj = businessMap.get(mainClass.getName());
                    if (obj == null) {
                        obj = mainClass.newInstance();
                        Method init = mainClass.getMethod("init");
                        init.invoke(obj);
                        businessMap.put(mainClass.getName(), obj);
                    }
                    BusinessResult busuinessRet = new BusinessResult();
                    Object[] args = new Object[] { busuinessRet };
                    Date startTime = new Date();
                    Method m = mainClass.getMethod("runBusiness", new Class[] { BusinessResult.class });
                    m.invoke(obj, args);

                    RemoteTaskService taskService = JDKProxy.createProxy(RemoteTaskService.class, "remote");
                    TaskLogBean log = convertResultToLog(startTime, busuinessRet, etask);
                    taskService.addTaskLog(log);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        //把运行结果，转成结果日志记录
        public TaskLogBean convertResultToLog(Date startTime, BusinessResult result, TaskExecutableBean etask) {
            TaskLogBean log = new TaskLogBean();
            log.setStartTime(startTime);
            log.setJobId(etask.getJobId());
            log.setTaskId(etask.getTaskId());
            log.setTaskGroup(etask.getTaskGroup());
            log.setTaskIdentity(etask.getTaskIdentity());
            log.setFlag(result.getAction().value());
            log.setMsg(result.getMsg());
            log.setLogType("zhaopin");
            log.setEndTime(new Date());
            return log;
        }
    }

}
